package com.atguigu.edu.realtime.dwd.db.app;

import com.atguigu.edu.realtime.common.base.BaseSQLApp;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * 考试域-测验事务事实表
 */
public class DwdTestExamDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdTestExamDetail().start(
                10001,
                4,
                Constant.TOPIC_DWD_TEST_EXAM_DETAIL
        );

    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        //TODO 1.设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(15*60 + 10));
        //TODO 2.从topic_db主题中读取数据  创建动态表
        readOdsDB(tableEnv,Constant.TOPIC_DWD_TEST_EXAM_DETAIL);

        // TODO: 2024/10/21 3.先过滤出测验表  test_exam
        Table testExamDetail = tableEnv.sqlQuery("select " +
                " `data`['score'] score, " +
                " `data`['deleted'] deleted, " +
                " `data`['create_time'] create_time, " +
                " `data`['user_id'] user_id, " +
                " `data`['submit_time'] submit_time, " +
                " `data`['id'] id, " +
                " `data`['paper_id'] paper_id, " +
                " `data`['duration_sec'] duration_sec, " +
                " `ts` " +
                "from topic_db " +
                "where `table`='test_exam' " +
                "and `type`='insert' ");
        //testExamDetail.execute().print();
        //tableEnv.createTemporaryView("dwd_test_exam_detail", testExamDetail);

        //TODO 8.将结果写到kafka主题
        //8.1 创建动态表和要写入的主题进行映射
        tableEnv.executeSql(
                "create table dwd_test_exam_detail(" +
                        "score string," +
                        "deleted string," +
                        "create_time string," +
                        "user_id string," +
                        "submit_time string," +
                        "id string," +
                        "paper_id string," +
                        "duration_sec string," +
                        "ts bigint," +
                        "primary key(id) not enforced " +
                        ")" + SQLUtil.getUpsertKafkaDDL(Constant.TOPIC_DWD_TEST_EXAM_DETAIL));

        //8.2 写入
        testExamDetail.executeInsert(Constant.TOPIC_DWD_TEST_EXAM_DETAIL);
    }
}
